package com.eshipenet.shipnet.equipmentdatahub
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.TimerService
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

import scala.concurrent.duration.FiniteDuration

case class IterableData(ed: Option[(EquipmentWithModel, EdgeChannelData)],
                        data: Option[EquipmentData],
                        delay: Option[FiniteDuration])

class IterableDataProcessor extends KeyedProcessFunction[String, IterableData, IterableData] {

  private lazy val desc =
    new ValueStateDescriptor[EquipmentDataState]("EquipmentDataState", classOf[EquipmentDataState])
  private lazy val state = getRuntimeContext.getState(desc)

  override def processElement(value: IterableData,
                              ctx: KeyedProcessFunction[String, IterableData, IterableData]#Context,
                              out: Collector[IterableData]): Unit = {
    if (value.ed.isEmpty) {
      out.close()
      return
    }
    val ed = value.ed.get
    val ds = Option(state.value()) match {
      case Some(s) => s
      case _       => new EquipmentDataState(ed._1, Map[String, EquipmentDataEntry](), None)
    }
    var nds = if (ds.isUpdated(ed._1)) {
      ds.resetWith(ed._1)
    } else {
      ds
    }
    nds = nds.input(ed._2)
    if (nds.shouldDelay.isDefined) {
      val it = value.copy(data = None, delay = nds.shouldDelay)
      out.collect(it)
    } else {
      scheduleOutput(ctx.timerService(), ctx.timerService().currentProcessingTime(), nds)
    }
    state.update(nds)
    out.close()
//    println("input", state.value())
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[String, IterableData, IterableData]#OnTimerContext,
                       out: Collector[IterableData]): Unit = {
    Option(state.value()) match {
      case Some(s) =>
        val data = s.output()
        state.update(data._1)
        out.collect(IterableData(None, Some(data._2), None))
        scheduleOutput(ctx.timerService(), timestamp, s)
      case _ => Unit
    }
    out.close()
  }

  private def scheduleOutput(ctx: TimerService, timestamp: Long, ds: EquipmentDataState): Unit = {
    val p = ds.equipment.model.period.getMillis
    val t = ((timestamp / p) + 1) * p
    ctx.registerProcessingTimeTimer(t)
  }

}
